package com.eshipenet.shipnet.equipmentdatahub

import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
import org.apache.flink.api.scala._
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema
import org.apache.kafka.clients.consumer.ConsumerRecord

class RawKafkaDeserializer() extends KafkaDeserializationSchema[RawKafkaMessage] {

  override def deserialize(record: ConsumerRecord[Array[Byte], Array[Byte]]): RawKafkaMessage = {
    val key   = new String(record.key(), "UTF-8")
    val value = new String(record.value(), "UTF-8")
    RawKafkaMessage(key, value)
  }

  override def isEndOfStream(nextElement: RawKafkaMessage): Boolean = false

  override def getProducedType: TypeInformation[RawKafkaMessage] = createTypeInformation[RawKafkaMessage]
}
